package com.shujia.spark

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo15Broadcast {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName("Demo15Broadcast")
      .setMaster("local")
      .set("spark.shuffle.file.buffer","64k")
    /**
      * spark-submit  --conf spark.shuffle.file.buffer=64k
      *
      */

    val sc = new SparkContext(conf)

    val studentRDD: RDD[String] = sc.textFile("spark/data/students.txt")

    //取出性别为男的学生
    studentRDD.filter(line => {
      val gender = line.split(",")(3)

      "男".equals(gender)
    })

    //clazzs   四Driver端定义的一个变量
    val clazzs = List("文科一班", "文科二班", "文科三班", "文科四班")

    //取出文科一班 二班 三班四班的学生
    studentRDD.filter(line => {
      val clazz = line.split(",")(4)
      //每一个task都会有一个变量副本
      clazzs.contains(clazz)
    })


    /**
      * 广播变量
      * 1、将变量在Driver广播出去
      * 2、在Executor端使用 在task中使用
      */

    //广播变量
    val clazzBro: Broadcast[List[String]] = sc.broadcast(clazzs)


    studentRDD.filter(line => {
      val clazz = line.split(",")(4)

      //获取广播变量
      /**
        * task在执行的过程中  如果使用到广播变量  task会向Executor获取
        * 如果Executor中没有这个广播变量 Executor会去Driver端拉去广播变量的数据
        *
        * 如果后面还有其他task使用到这个广播变量  就不需要再拉去了
        * 所以每一个Executor中只会有一个变量副本
        *
        * 广播变量不能再Executor端修改
        *
        */
      val clazzs: List[String] = clazzBro.value

      clazzs.contains(clazz)
    })


    /**
      * 累加器
      *
      * 在Executor中进行局部累加   当job执行完成  在到Driver进行汇总
      *
      */

    //在Driver 创建 累加器
    val acc = sc.accumulator(0)

    studentRDD.foreach(line => {
      //在Executor端累加
      acc.add(1)

    })

    //最后在Driver端读取累加器的结果
    println(acc.value)


  }
}
